package com.hexinfo.dmpro.data.disaster.recovery.backups.handler.impl;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.StrUtil;
import com.hexinfo.dmpro.component.common.service.impl.HiveQueryService;
import com.hexinfo.dmpro.component.common.util.DataSourceUtil;
import com.hexinfo.dmpro.component.comparison.entity.TableContrastParamEntity;
import com.hexinfo.dmpro.component.comparison.entity.TableContrastResultEntity;
import com.hexinfo.dmpro.component.comparison.enums.TableContrastEnums;
import com.hexinfo.dmpro.component.comparison.service.ITableContrastService;
import com.hexinfo.dmpro.component.data.transfer.model.DataExport;
import com.hexinfo.dmpro.component.data.transfer.model.DataImport;
import com.hexinfo.dmpro.component.data.transfer.service.DataExportService;
import com.hexinfo.dmpro.component.data.transfer.service.DataImportService;
import com.hexinfo.dmpro.component.file.transfer.model.FileTransfer;
import com.hexinfo.dmpro.component.file.transfer.service.FileTransferService;
import com.hexinfo.dmpro.data.disaster.recovery.backups.handler.IDataDisasterRecoveryHandler;
import com.hexinfo.dmpro.data.disaster.recovery.backups.model.DisasterRecoveryParam;
import com.hexinfo.dmpro.data.disaster.recovery.common.constants.FtpConnConstants;
import com.hexinfo.dmpro.data.disaster.recovery.common.util.TarGzUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.Connection;

@Slf4j
@Service
public class DataDisasterRecoveryHandler implements IDataDisasterRecoveryHandler {

    @Autowired
    private DataExportService dataExportService;

    @Autowired
    private DataImportService dataImportService;

    @Autowired
    private FileTransferService fileTransferService;

    @Autowired
    private FtpConnConstants ftpConnConstants;

    @Autowired
    private ITableContrastService tableContrastService;

    @Autowired
    private HiveQueryService hiveQueryService;

    @Override
    public void disasterRecovery(DisasterRecoveryParam disasterRecoveryParam) {
        try {
            // ① 数据导出(HIVE数据生成TXT文本)
            dataExportService.dataExportToLocal(buildDataExport(disasterRecoveryParam));
            // ② 数据备份(将生成的TXT文本导入到HIVE备库中)
            dataImportService.dataImportFromLocal(buildDataImportParam(disasterRecoveryParam));
            // ③ 数据文件压缩(TXT数据文件+DDL建表语句)
            dataFileCompress(disasterRecoveryParam);
            // ④ 数据文件上传到离线集群（离线集群近一年内的数据做快照、数据文件上传、本地数据文件删除）
            dataFileUpload(disasterRecoveryParam);
            // ⑤ 离线集群数据文件下载（下载、解压）
            dataFileDown(disasterRecoveryParam);
            // ⑦ 数据文件导入（TXT数据文件导入到验库中、本地数据文件删除）
            dataFileImport(disasterRecoveryParam);
            // ⑧ 数据比对
            dataCompare(disasterRecoveryParam);
            // ⑨ 数据清理
            dataClean(disasterRecoveryParam);
        } catch (Exception e) {
            log.error("数据灾难恢复异常", e);
        }
    }

    private DataExport buildDataExport(DisasterRecoveryParam disasterRecoveryParam) {
        DataExport dataExport = new DataExport();
        dataExport.setHdfsUrls(disasterRecoveryParam.getHdfsUrls());
//        dataExport.setHdfsPath(dataExport.getHdfsPath());
        dataExport.setLocalDir(disasterRecoveryParam.getLocalDir());
        dataExport.setLocalFileName(disasterRecoveryParam.getFileName());
        dataExport.setHiveQuerySql(disasterRecoveryParam.getHiveQuerySql());
        dataExport.setDatabaseName(disasterRecoveryParam.getDatabaseName());
        dataExport.setTableName(disasterRecoveryParam.getTableName());
        dataExport.setSeparator(disasterRecoveryParam.getSeparator());
        return dataExport;
    }

    private DataImport buildDataImportParam(DisasterRecoveryParam disasterRecoveryParam) {
        DataImport dataImport = new DataImport();
        dataImport.setHdfsUrls(disasterRecoveryParam.getHdfsUrls());
//        dataImport.setHdfsPath(disasterRecoveryParam.getImportHdrsPath());
        dataImport.setHdfsFileName(disasterRecoveryParam.getFileName());
        dataImport.setLocalDir(disasterRecoveryParam.getLocalDir());
        dataImport.setLocalFileName(disasterRecoveryParam.getFileName());
        dataImport.setImportType(disasterRecoveryParam.getImportType());
        dataImport.setDatabaseName(disasterRecoveryParam.getDatabaseName());
        dataImport.setTableName(disasterRecoveryParam.getTableName());
//        dataImport.setIsAutoCreateTable("N");
        return dataImport;
    }

    private void dataFileCompress(DisasterRecoveryParam disasterRecoveryParam) {
        try {
            // ① 数据文件TXT + 表DDL文件压缩
            TarGzUtil.compress(disasterRecoveryParam.getBackupFiles(), disasterRecoveryParam.getBackupTarGzFilePath());
            // ② 数据文件TXT删除
            FileUtil.del(disasterRecoveryParam.getExportLocalPath());
        } catch (Exception e) {
            log.error("[生成数据冷备] {} 数据文件压缩失败: {}", disasterRecoveryParam.getFileName(), e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("数据文件压缩失败: {}", e.getMessage()));
        }
    }

    private void dataFileUpload(DisasterRecoveryParam disasterRecoveryParam) {
        FileTransfer fileTransfer = buildUploadFileTransfer(disasterRecoveryParam);
        // ① 离线集群近一年数据文件做快照
        fileTransferService.fileTransfer(fileTransfer);
        // ② 数据文件上传到离线集群
        fileTransferService.fileUpload(fileTransfer);
        // ③ 本地数据文件.tar.gz删除
        FileUtil.del(disasterRecoveryParam.getBackupTarGzFilePath());
    }

    private FileTransfer buildUploadFileTransfer(DisasterRecoveryParam disasterRecoveryParam) {
        FileTransfer fileTransfer = new FileTransfer();
        fileTransfer.setTransferType("FTP");
        fileTransfer.setServer(ftpConnConstants.getServer());
        fileTransfer.setPort(ftpConnConstants.getPort());
        fileTransfer.setUser(ftpConnConstants.getUser());
        fileTransfer.setPassword(ftpConnConstants.getPassword());
        fileTransfer.setLocalDir(disasterRecoveryParam.getBackupTarGzFileDir());
        fileTransfer.setRemoteDir(disasterRecoveryParam.getRemoteFilePath());
        fileTransfer.setFileName(disasterRecoveryParam.getTarGzFileName());
        fileTransfer.setRemoteFileNewPath(disasterRecoveryParam.getRemoteBackUpFilePath());
        fileTransfer.setRemoteFileNewName(disasterRecoveryParam.getFileName());
        return fileTransfer;
    }

    private void dataFileDown(DisasterRecoveryParam disasterRecoveryParam) {
        try {
            FileTransfer fileTransfer = buildDownFileTransfer(disasterRecoveryParam);
            // ① 数据文件下载
            fileTransferService.fileDownload(fileTransfer);
            // ② 数据文件解压
            TarGzUtil.decompress(fileTransfer.getLocalFilePath(), disasterRecoveryParam.getExportLocalDir());
            // ② 数据文件TXT删除
            FileUtil.del(fileTransfer.getLocalFilePath());
        } catch (Exception e) {
            log.error("[生成数据冷备] {} 数据文件解压失败: {}", disasterRecoveryParam.getFileName(), e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("数据文件解压失败: {}", e.getMessage()));
        }
    }

    private FileTransfer buildDownFileTransfer(DisasterRecoveryParam disasterRecoveryParam) {
        FileTransfer fileTransfer = new FileTransfer();
        fileTransfer.setTransferType("FTP");
        fileTransfer.setServer(ftpConnConstants.getServer());
        fileTransfer.setPort(ftpConnConstants.getPort());
        fileTransfer.setUser(ftpConnConstants.getUser());
        fileTransfer.setPassword(ftpConnConstants.getPassword());
        fileTransfer.setLocalDir(disasterRecoveryParam.getBackupTarGzFileDir());
        fileTransfer.setRemoteDir(disasterRecoveryParam.getRemoteFilePath());
        fileTransfer.setFileName(disasterRecoveryParam.getTarGzFileName());
        fileTransfer.setRemoteFileNewPath(disasterRecoveryParam.getRemoteBackUpFilePath());
        fileTransfer.setRemoteFileNewName(disasterRecoveryParam.getFileName());
        return fileTransfer;
    }

    private void dataFileImport(DisasterRecoveryParam disasterRecoveryParam) {
        // ① 数据文件导入到验库中
        dataImportService.dataImportFromLocal(buildDataImportParam(disasterRecoveryParam));
        // ② 删除数据文件
        FileUtil.del(disasterRecoveryParam.getExportLocalDir());
    }

    private void dataCompare(DisasterRecoveryParam disasterRecoveryParam) {
        TableContrastParamEntity tableContrastParam = buildTableContrastParam(disasterRecoveryParam);
        tableContrastService.createTableContrastTask(tableContrastParam);
        boolean isTableDataCompareFinish = false;
        while (!isTableDataCompareFinish) {
            TableContrastResultEntity tableContrastResultEntity = tableContrastService.queryTableContrastResult(tableContrastParam.getTableFlag());
            if (TableContrastEnums.ACHIEVE.getValue().equals(tableContrastResultEntity.getStatus())) {
                isTableDataCompareFinish = true;
            } else if (TableContrastEnums.FAIL.getValue().equals(tableContrastResultEntity.getStatus())) {
                isTableDataCompareFinish = true;
            }
        }
    }

    private TableContrastParamEntity buildTableContrastParam(DisasterRecoveryParam disasterRecoveryParam) {
        TableContrastParamEntity tableContrastParam = new TableContrastParamEntity();
        tableContrastParam.setContrastType("");
        tableContrastParam.setSourceDbName("");
        tableContrastParam.setSourceTableName("");
        tableContrastParam.setSourceQueryCriteria("");
        tableContrastParam.setTargetDbName("");
        tableContrastParam.setTargetTableName(disasterRecoveryParam.getTableName());
        tableContrastParam.setTargetQueryCriteria("");
        return tableContrastParam;
    }

    private void dataClean(DisasterRecoveryParam disasterRecoveryParam) {
        Connection connection = DataSourceUtil.getHiveTargetConnection();
        if (StrUtil.isNotBlank(disasterRecoveryParam.getPartitionName())) {
            hiveQueryService.delTablePartition(connection, "", "", "", "");
            hiveQueryService.delTablePartition(connection, "", "", "", "");
        } else {
            hiveQueryService.dropTable(connection, "", "");
            hiveQueryService.dropTable(connection, "", "");
        }
    }
}
